package async;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.x.async.AsyncCuratorFramework;
import org.apache.curator.x.async.AsyncEventException;
import org.apache.curator.x.async.WatchMode;
import org.apache.zookeeper.WatchedEvent;

import com.zhb.cloud.curator.utils.CuratorFrameworkUtils;

import java.util.concurrent.CompletionStage;

/**
 * @ClassName: AsyncExamples
 * @author: zhb
 * @date: 2017年11月5日 上午10:36:50
 * @Description: 使用异步DSL的例子
 */
@SuppressWarnings("all")
public class AsyncExamples {

	/**
	 * NAME_SPACE
	 */
	private static final String PATH = "zhb/zk";
	/**
	 * zookeeper hosts address
	 */
	private static final String PAYLOAD = "测试异步DSL调用";

	/**
	 * @Title: wrap
	 * @author: zhb
	 * @date: 2017年11月5日 上午10:39:41
	 * @Description: 包装CuratorFramework实例，以便可以使用异步。
	 *               这样做一次并重新使用返回的AsyncCuratorFramework实例
	 * @param client
	 * @return
	 */
	public static AsyncCuratorFramework wrap(CuratorFramework client) {
		return AsyncCuratorFramework.wrap(client);
	}

	/**
	 * @Title: create
	 * @author: zhb
	 * @date: 2017年11月5日 上午10:40:16
	 * @Description: 获得一个 CuratorFramework客户端
	 * @param client
	 * @param path
	 * @param payload
	 */
	public static void create(CuratorFramework client, String path, byte[] payload) {
		AsyncCuratorFramework async = AsyncCuratorFramework.wrap(client);
		/**
		 * 使用给定的有效负载异步地在给定的路径上创建一个节点
		 */
		async.create().forPath(path, payload).whenComplete((name, exception) -> {
			if (exception != null) {
				// there was a problem
				exception.printStackTrace();
			} else {
				System.out.println("Created node name is: " + name);
			}
		});
	}

	/**
	 * @Title: createThenWatch
	 * @author: zhb
	 * @date: 2017年11月5日 上午10:43:05
	 * @Description: 这个例子显示了为这两个事件异步使用观察者触发和连接问题。 如果你不需要通知 连接问题，请使用更简单的方法
	 * @param client
	 * @param path
	 */
	public static void createThenWatch(CuratorFramework client, String path) {
		AsyncCuratorFramework async = AsyncCuratorFramework.wrap(client);
		// 以给定的有效载荷异步地在给定路径上创建一个节点
		// 然后观看创建的节点
		async.create().forPath(path).whenComplete((name, exception) -> {
			if (exception != null) {
				exception.printStackTrace();
			} else {
				handleWatchedStage(async.watched().checkExists().forPath(path).event());
			}
		});
	}

	/**
	 * @Title: createThenWatchSimple
	 * @author: zhb
	 * @date: 2017年11月5日 上午10:43:54
	 * @Description:
	 * @param client
	 * @param path
	 */
	public static void createThenWatchSimple(CuratorFramework client, String path) {
		// 以给定的有效载荷异步地在给定路径上创建一个节点 然后观看创建的节点
		AsyncCuratorFramework async = AsyncCuratorFramework.wrap(client);
		async.create().forPath(path).whenComplete((name, exception) -> {
			if (exception != null) {
				exception.printStackTrace();
			} else {
				// because "WatchMode.successOnly" is used the watch stage is
				// only triggered when
				// the EventType is a node event
				// 因为“WatchMode.successOnly”用于watch阶段,只有在触发时 EventType是一个节点事件
				async.with(WatchMode.successOnly).watched().checkExists().forPath(path).event().thenAccept(event -> {
					System.out.println(event.getType());
					System.out.println(event);
				});
			}
		});
	}

	/**
	 * @Title: handleWatchedStage
	 * @author: zhb
	 * @date: 2017年11月5日 上午10:47:30
	 * @Description: 处理监听状态
	 * @param watchedStage
	 */
	private static void handleWatchedStage(CompletionStage<WatchedEvent> watchedStage) {
		// async handling of Watchers is complicated because watchers can
		// trigger multiple times
		// and CompletionStage don't support this behavior
		// 观察者的异步处理很复杂，因为观察者可以多次触发并且CompletionStage不支持这种行为
		// thenAccept() handles normal watcher triggering.
		// thenAccept()处理正常的观察者触发。
		watchedStage.thenAccept(event -> {
			System.out.println(event.getType());
			System.out.println(event);
		});

		// exceptionally is called if there is a connection problem in which
		// case
		// watchers trigger to signal the connection problem. "reset()" must be
		// called
		// to reset the watched stage
		// 如果在其中存在连接问题，则异常地被调用案件观察者触发以发出连接问题的信号。 “reset（）”必须被调用重置监控状态
		watchedStage.exceptionally(exception -> {
			AsyncEventException asyncEx = (AsyncEventException) exception;
			asyncEx.printStackTrace(); // handle the error as needed
			handleWatchedStage(asyncEx.reset());
			return null;
		});
	}

	public static void main(String[] args) {
		CuratorFramework framework = CuratorFrameworkUtils.initCuratorFramework();
		AsyncCuratorFramework asyncCuratorFramework = wrap(framework);
		create(framework, PATH, PAYLOAD.getBytes());
	}
}
